package com.link.base.base.kafkalisteners;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.link.base.base.accnt.model.Account;
import com.link.base.base.accnt.service.AccountService;
import com.link.base.base.enterprise.service.EnterpriseService;
import com.link.base.base.user.service.UserService;
import com.link.base.core.basic.interceptors.mybatis.system.KafkaSender;
import com.link.base.microinterface.intercust.microservice.InterCustClient;
import com.link.core.basic.service.KeyGenerateService;
import com.link.core.cllog.LCLogger;
import com.link.core.kafka.KafkaConnectCondition;
import com.link.core.util.DateUtil;
import com.link.core.util.JsonUtil;
import com.link.core.util.StringUtils;
import com.link.core.util.TokenUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.eclipse.jetty.util.StringUtil;
import org.springframework.context.annotation.Conditional;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author yaojiangyi
 * @description basekafka监听类 消费来源base的消息
 * @date 2019/07/15 11:38
 */
@Component
@Conditional(KafkaConnectCondition.class)
public class BaseKafkaListeners {
    private static final Logger logger = LogManager.getLogger(BaseKafkaListeners.class);
    private static final String SUCCESS = "success";
    private static final String MEMBER_PLATFORM = "MemberPlatform";
    private static final String MARKETING_PLATFORM = "MarketingPlatform";
    /**
     * 订单调用标识
     */
    private static final String SALE_ORDER_PLATFORM = "SaleOrderPlatform";


    @Resource
    private AccountService accountService;

    @Resource
    private InterCustClient interCustClient;
    @Resource
    protected KeyGenerateService keyGenerateService;

    @Resource
    private UserService userService;

    @Resource
    private EnterpriseService enterpriseService;

    /**
     * 监控kafka消息 获取渠道消费者 目前只有营销平台与会员平台需要使用
     *
     * @author yaojiangyi
     * @date 2019-04-28 17:15
     */
    @KafkaListener(topics = {"mc_accntchannel_topic"})
    public void obtainAccntChannelId(List<ConsumerRecord> records) throws Exception {
        for (ConsumerRecord record : records) {
            if (record.value() != null) {
                String jsonString = record.value().toString();
                JSONObject jsonObject = JSONObject.parseObject(jsonString);
                String accntChannelType = jsonObject.get("acctChannelType").toString();
                // MarketingPlatform为营销平台生成的渠道消费者 MemberPlatform为会员平台生成的渠道消费者
                if (MARKETING_PLATFORM.equals(accntChannelType)) {
                    marketPlatChannelId(jsonObject);
                }
//                else if (MEMBER_PLATFORM.equals(accntChannelType)) {
//                    memberAcctUpdate(jsonObject);
//                } else if (SALE_ORDER_PLATFORM.equals(accntChannelType)) {
//                    saleOrderAcctId(jsonObject);
            } else {
                logger.warn("mc_accntchannel_topic：传入平台类型异常！");
            }
        }
    }

    /**
     * 根据营销活动获取人群对应的渠道消费者id
     *
     * @author yaojiangyi
     * @date 2019-06-20 16:54
     */
    private void marketPlatChannelId(JSONObject jsonObject) throws Exception {
        // 活动人群id
        String interCustId = jsonObject.get("in_cust_id").toString();
        String userName = jsonObject.get("user_name").toString();
        TokenUtils.setLogin(userName);

        // 渠道消费者id
        String accntChannelId = jsonObject.get("accnt_channel_id").toString();
        try {
            String errMsg = "";
            if (StringUtil.isBlank(interCustId)) {
                errMsg += "mc_accntchannel_topic_mapl：未收到活动人群id";
                logger.error(errMsg);
                return;
            }
            Account channelAccnt = JSON.parseObject(JSON.toJSONString(jsonObject), Account.class);

            // 如果原记录无渠道消费者id才更新
            if (accntChannelId.equals("0")) {

                Account accnt = new Account();
                accnt.setMobilePhone1(jsonObject.get("mobilePhone1").toString());
                accnt.setAcctChannel(MARKETING_PLATFORM);
                List<Account> channelAccnts = accountService.queryByExamplePage(accnt);
                Long channelAcctId;
                if (channelAccnts.isEmpty()) {
                    channelAccnt.setId(keyGenerateService.keyGenerate());
                    channelAccnt.setAcctStatus("Y");
                    Date nowDt = new Date();
                    String nowDtStr = DateUtil.dateToStr(nowDt, "yyyy-MM-dd HH:mm:ss");
                    //channelAccnt.setFirstEngageTime(nowDtStr);
                    accountService.insert(channelAccnt);
                    accnt = accountService.queryById(channelAccnt);
                    channelAcctId = accnt.getId();
                    Object object = userService.transformToDmpConsumer(accnt);
                    // 把消费者同步到dmp
                    KafkaSender.sendInfo("CDC_CONSUMER", JsonUtil.toString(object));
                } else {
                    channelAcctId = channelAccnts.get(0).getId();
                }
                JSONObject jsonCustObject = new JSONObject();
                jsonCustObject.put("id", interCustId);
                jsonCustObject.put("accntChannelId", channelAcctId);     // 渠道消费者id

                Map<String, Object> mapResult = new HashMap<>();
                try {
                    mapResult = interCustClient.updateChannelMsg(jsonCustObject);
                    Boolean successFlg = (Boolean) mapResult.get("success");
                    if (!successFlg) {
                        errMsg += " ### mc_accntchannel_topic_mapl【更新营销活动】报错:" + mapResult.get("result").toString();
                        logger.error(errMsg);
                    }
                } catch (Exception e) {
                    errMsg += " ### Exception-->mc_accntchannel_topic_mapl【更新营销活动】报错:" + e.getMessage();
                    logger.error(errMsg);
                    LCLogger.withException(e);
                }
                // 记录错误信息
                if (StringUtils.isNotBlank(errMsg)) {
                    LCLogger.error().withMessageKey("BaseKafkaListeners")
                            .withMessage("调用营销平台更新营销活动错误: " + errMsg).flush();
                }
            }
        } catch (Exception e) {
            LCLogger.error().withMessageKey("BaseKafkaListeners").withMessage("kafka更新渠道消费者报错: " + e.getMessage()).flush();
            LCLogger.withException(e);
        }


    }
}
